package com.xhwl.centiotdata.emq.dnake;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.xhwl.centiotdata.pojo.enums.talk.ModeEnum;
import com.xhwl.centiotdata.service.cent.ICentTalkBackService;
import com.xhwl.centiotdata.utils.RedisUtils;
import com.xhwl.common.enums.talkback.CallTypeEnum;
import com.xhwl.common.exception.BusinessException;
import com.xhwl.common.pojo.cent.talkback.TalkRecord;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import javax.annotation.PostConstruct;
import java.text.SimpleDateFormat;
import java.util.*;

import static com.xhwl.centiotdata.pojo.constant.DnakeConstant.*;

/**
 * @description:
 * @author: wuyuhao
 * @createDate: 2023/10/7 10:40
 */
@Component
@Configuration
@ConfigurationProperties(DnakeMqttClient.PREFIX)
@Slf4j
public class DnakeMqttClient implements MqttCallbackExtended {

    /**
     * 指定配置文件application-local.properties中的属性名前缀
     */
    public static final String PREFIX = "mqtt.dnake.producer";

    @Setter
    @Getter
    private String host;
    @Setter
    @Getter
    private String clientId;
    @Setter
    @Getter
    private String username;
    @Setter
    @Getter
    private String password;
    @Setter
    @Getter
    private int timeout;
    @Setter
    @Getter
    private List<String> topic;

    private IMqttClient client;

    @Autowired
    ICentTalkBackService talkBackService;

    @Autowired
    RedisUtils redisUtils;

    /**
     * 初始化mqtt客户端
     */
    @PostConstruct
    public void initMqttClient() {
        try {
            client = new MqttClient(host, clientId, new MemoryPersistence());
            client.setCallback(this);

            initConnect(username, password, timeout);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * mqtt客户端初始化连接
     *
     * @param username
     * @param password
     * @param timeout
     * @throws MqttException
     */
    public void initConnect(String username, String password, int timeout) {
        MqttConnectOptions options = new MqttConnectOptions();
        // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
        options.setCleanSession(true);
        // 设置连接的用户名
        if (username != null) {
            options.setUserName(username);
        }
        // 设置连接的密码
        if (password != null) {
            options.setPassword(password.toCharArray());
        }
        // 设置超时时间 单位为秒
        options.setConnectionTimeout(timeout);
        // 设置会话心跳时间 单位为秒 服务器会每隔1.5*keepalive的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
        options.setKeepAliveInterval(20);
        options.setAutomaticReconnect(true);

        IMqttToken mqttToken;
        try {
            mqttToken = client.connectWithResult(options);
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
        if (mqttToken.isComplete()) {
            log.info("MQTT连接成功");
            subscribeTopic();
        } else {
            log.error("MQTT连接失败, msg: {}", mqttToken.getException().toString());
        }
    }

    /**
     * mqtt客户端订阅topic
     *
     * @throws MqttException
     */
    public void subscribeTopic() {
        log.info("订阅主题列表: {}", topic);
        try {
            client.subscribe(topic.toArray(new String[0]), new int[]{1});
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 发布消息
     *
     * @param topic
     * @param message
     */
    public void publish(String topic, String message) {
        log.info("mqtt发布消息, topic: {}, message: {}", topic, message);
        MqttMessage mqttMessage = new MqttMessage();
        mqttMessage.setQos(0);
        // 非持久化
        mqttMessage.setRetained(false);
        mqttMessage.setPayload(message.getBytes());
        if (client == null) {
            return;
        }
        MqttTopic mqttTopic = client.getTopic(topic);
        if (null == mqttTopic) {
            throw new BusinessException("主题不存在");
        }
        try {
            mqttTopic.publish(mqttMessage);
        } catch (Exception e) {
            log.error("mqtt发送消息异常, msg: {}", e.toString());
        }
    }

    /**
     * 断开连接
     */
    public void disconnect() {
        if (client != null) {
            try {
                client.disconnect();
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }
    }

    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        log.info("mqtt客户端连接成功");
        if (reconnect) {
            log.info("mqtt客户端重新订阅topic");
            subscribeTopic();
        }
    }

    @Override
    public void connectionLost(Throwable cause) {
        log.error("mqtt客户端断开连接, msg: {}", cause.toString());
    }

    @Override
    @Async("asyncServiceExecutor")
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        String payload = new String(message.getPayload());
        log.info("消息订阅, topic: {}, messageId: {}, qos: {}, payload: {}",
                topic, message.getId(), message.getQos(), payload);
        JSONObject object = JSON.parseObject(payload);
        if (ObjectUtils.nullSafeEquals(DEVICE_TOPICS, topic)) {
            deviceTopic(object);
        } else if (ObjectUtils.nullSafeEquals(String.format(RESPOND, "xhwl"), topic)) {
            respond(object);
        }
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        int messageId = token.getMessageId();
        String[] topics = token.getTopics();
        log.info("消息发布完成, messageId: {}, topics: {}", messageId, topics);
    }

    private void respond(JSONObject object) {
        String action = object.getString("action");
        if (ObjectUtils.nullSafeEquals(action, "remoteOpen")) {
            log.info("狄耐克设备远程开门下发结果: {}", object);
        }
    }

    private void deviceTopic(JSONObject object) {
        String action = object.getString("action");
        if (ObjectUtils.nullSafeEquals(action, "deviceInfo")) {
            JSONObject params = object.getJSONObject("params");
            String itemNumber = params.getString("itemNumber");
            if (ObjectUtils.isEmpty(itemNumber)) {
                return;
            }
            String[] split = itemNumber.split("-");
            String projectId = split[0];
            // 默认第一个是项目id
            if (!ObjectUtils.isEmpty(projectId)) {
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                String now = sdf.format(new Date());
                params.put("time", now);
                log.info("deviceInfo, 狄耐克设备数据: {}", params);
                String uid = params.getString("uid");
                String key = String.format(DEVICE_INFO_REDIS_KEY, projectId, uid);
                redisUtils.set(key, params);
            }
        } else if (ObjectUtils.nullSafeEquals(action, "heartSync")) {
            String uid = object.getString("from");
            JSONObject params = object.getJSONObject("params");
            String key = String.format(DEVICE_INFO_REDIS_KEY, "*", uid);
            List<Object> values = redisUtils.getValues(key);
            Optional<Object> optional = values.stream()
                    .max(Comparator.comparing(e -> ((JSONObject) e).getString("time")));
            if (optional.isPresent()) {
                JSONObject jsonObject = (JSONObject) optional.get();
                // 当前正在心跳的设备
                log.info("heartSync, 狄耐克设备数据: {}", jsonObject);
                String itemNumber = jsonObject.getString("itemNumber");
                if (ObjectUtils.isEmpty(itemNumber)) {
                    return;
                }
                String[] split = itemNumber.split("-");
                String projectId = split[0];
                String k = String.format(DEVICE_ONLINE_REDIS_KEY, projectId, uid);
                redisUtils.setWithSeconds(k, new JSONObject(), 15);
                // 设备信息加上ip
                jsonObject.put("ip", params.get("ip"));
                redisUtils.set(String.format(DEVICE_INFO_REDIS_KEY, projectId, uid), jsonObject);
            }
        } else if (ObjectUtils.nullSafeEquals(action, "callRecord")) {// 呼叫记录
            JSONObject params = object.getJSONObject("params");
            log.info("callRecord, 狄耐克对讲记录: {}", params);
            Integer mode = params.getInteger("mode");
            if (mode.equals(ModeEnum.ANSWERED.getCode()) || mode.equals(ModeEnum.NOT_ANSWERED.getCode())) {
                String callNum = params.getString("callNum");
                // 保存在云对讲的数据库中
                TalkRecord talkRecord = new TalkRecord();
                talkRecord.setCallerId(callNum);
                String uid = object.getString("from");
                String key = String.format(DEVICE_INFO_REDIS_KEY, "*", uid);
                List<Object> values = redisUtils.getValues(key);
                Optional<Object> optional = values.stream()
                        .max(Comparator.comparing(e -> ((JSONObject) e).getString("time")));
                if (optional.isPresent()) {
                    JSONObject value = (JSONObject) optional.get();
                    String deviceType = value.getString("deviceType");
                    if ("0101".equals(deviceType) || "0102".equals(deviceType)) {
                        talkRecord.setCallRoute(CallTypeEnum.UNIT_DEVICE);
                    } else if ("0103".equals(deviceType)) {
                        talkRecord.setCallRoute(CallTypeEnum.FENCE_DEVICE);
                    } else if ("0104".equals(deviceType)) {
                        talkRecord.setCallRoute(CallTypeEnum.MANAGE_DEVICE);
                    }
                    talkRecord.setAnswer(mode.equals(ModeEnum.ANSWERED.getCode()));
                    talkBackService.saveTalkRecordList(Collections.singletonList(talkRecord));
                }
            }
        }
    }

}
